package com.infinitus.autocatlog.task;

import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;
import com.infinitus.autocatlog.domain.FilterCountLog;
import com.infinitus.autocatlog.domain.FilterLog;
import com.infinitus.autocatlog.domain.KafkaConfig;
import com.infinitus.autocatlog.domain.KafkaStartupLog;
import com.infinitus.autocatlog.domain.MatcherConfig;
import com.infinitus.autocatlog.domain.RecordItem;
import com.infinitus.autocatlog.service.KafkaLogService;
import com.infinitus.autocatlog.service.RedisService;
import com.infinitus.autocatlog.service.util.Base64Coder;
import com.infinitus.autocatlog.service.util.TimeUtil;


@Component
public class Cleaner  {

	private static final Logger LOG = LoggerFactory.getLogger(Cleaner.class);

	@Autowired
	private KafkaLogService kafkaLogService;
	
	@Autowired
	private RedisService redisService;
	
	@Autowired
	private LoadDataTask loadDataTask;
	
	@Autowired
	private KafkaConfig kafkaConfig;
	
	@Autowired
	private MatcherConfig matcherConfig;
	/**
	 * 清洗用户行为数据
	 * @throws InterruptedException 
	 */
	@Async
	public void userbehaviorClean(String topic, String osplat, boolean isPro){
		
		LOG.info("启动kafka消费端接收appPro数据");
		KafkaConsumer<String, String> consumer = createConsumer(topic);
		LOG.info("启动kafka生产者发送appPro数据");
		Properties props = kafkaConfig.getProperties(false);
		Producer<String, String> producer = new KafkaProducer<String, String>(props);
		
		int totalCount = 0;
		int errorCount = 0;
		String today = TimeUtil.getCurrentDay();
		int currentHour = TimeUtil.getCurrentHour();
		String today_total_key = MatcherConfig.TOTAL_COUNT + topic+ today;
		String today_error_key =MatcherConfig.ERROR_COUNT + topic+ today;
		if(redisService.exists(today_total_key)){
			totalCount = Integer.parseInt(redisService.get(today_total_key));
			if(redisService.exists(today_error_key)){
				errorCount = Integer.parseInt(redisService.get(today_error_key));
			}
		}else{
			redisService.set(today_total_key, totalCount + "");
			redisService.set(today_error_key, errorCount + "");
		}
		
		CloseableHttpClient httpclient = HttpClients.createDefault();
		try{
			//订阅消息
			ConsumerRecords<String, String> records = null;
			while (true) {
				try {
					Thread.sleep(10);
				} catch (InterruptedException e) {
					LOG.warn("sleep fail ,Continuing to pull data may cause memory overflow"+e);
				}
				records = null;
				try{
					records = consumer.poll(100);
					for (ConsumerRecord<String, String> record : records) {
						String value = record.value();
						String nginxInfo = value.substring(0, value.lastIndexOf("\t")+1);
						Map<String, String> fieldTypes = new HashMap<String, String>();
						List<RecordItem> items = RecordItem.converts(value, fieldTypes, topic);
						List<RecordItem> buff  = new ArrayList<RecordItem>();
						for(RecordItem one: items){
							//记录总数
							totalCount ++;
							if(totalCount>0 && totalCount%10==0){
								String crossKey = null;
								//检查是不是跨天了
								if(!today.equals(TimeUtil.getCurrentDay())){
									//更新昨天最后接收记录
									kafkaLogService.saveCountLog(new FilterCountLog( Integer.parseInt(redisService.get(today_total_key)), Integer.parseInt(redisService.get(today_error_key)), TimeUtil.dateFormat.parse(today), new Date(), topic));
									//更新昨天23时的过滤详细记录
									crossKey = topic+MatcherConfig.SEPARATOR+osplat+MatcherConfig.SEPARATOR + today +MatcherConfig.SEPARATOR+currentHour+"*";
									totalCount = 1;
									errorCount = 0;
									today = TimeUtil.getCurrentDay();
									today_total_key = MatcherConfig.TOTAL_COUNT +topic+ today;
									today_error_key = MatcherConfig.ERROR_COUNT +topic+ today;
								}else if(currentHour!=TimeUtil.getCurrentHour()){
									//没有跨天，只是跨小时，保存前一小时的完整记录
									crossKey = topic+MatcherConfig.SEPARATOR+osplat+MatcherConfig.SEPARATOR + today +MatcherConfig.SEPARATOR+currentHour+"*";
								}
								//更新上一个小时的过滤记录
								if(crossKey!=null){
									Set<Serializable> logKeys = redisService.keys(crossKey);
									for(Serializable s : logKeys ){
										String[] key = ((String) s).split(MatcherConfig.SEPARATOR);
										kafkaLogService.saveFilterLog(new FilterLog(topic, osplat, key[2], key[3], key[4], key[5], key[6],redisService.get((String)s) ,new Date()));
									}
									//更新小时
									currentHour = TimeUtil.getCurrentHour();
								}
								redisService.set(today_total_key, totalCount + "");
								//避免频繁写库
								if(totalCount%100 == 0){
									Date now = new Date();
									kafkaLogService.saveCountLog(new FilterCountLog(totalCount, errorCount, now, now, topic));
									if(totalCount%1000 == 0){
										Set<Serializable> logKeys = redisService.keys(topic+MatcherConfig.SEPARATOR+osplat+MatcherConfig.SEPARATOR + today +MatcherConfig.SEPARATOR+currentHour+"*");
										for(Serializable s : logKeys ){
											String[] key = ((String) s).split(MatcherConfig.SEPARATOR);
											kafkaLogService.saveFilterLog(new FilterLog(topic, osplat,  key[2], key[3], key[4], key[5], key[6],redisService.get((String)s) ,new Date()));
										}
									}
								}
							}
							//根据规则清洗
							String checkResult = FilterRules.checkAll(one, osplat);
							String logKey =topic+MatcherConfig.SEPARATOR+osplat+MatcherConfig.SEPARATOR + today +MatcherConfig.SEPARATOR + TimeUtil.getCurrentHour() + MatcherConfig.SEPARATOR + one.getEventDate() + MatcherConfig.SEPARATOR+one.getApp_name()+ MatcherConfig.SEPARATOR;
							//符合的记录进一步处理
							if(checkResult == null){
								dataHandle(httpclient, one, osplat);
								buff.add(one);
								logKey+="normal";
								/*if(isPro){
									producer.send(new ProducerRecord<String, String>(topic +"_etl", one.toString()));
								}*/
							}else{
								//不符合的记录日志
								errorCount ++;
								redisService.set(today_error_key, errorCount + "");
								logKey+=checkResult;
							}
							one.setOsplat(osplat+isPro);
							one.setErrorType(checkResult);
							kafkaLogService.saveLog(one);
							if(redisService.exists(logKey)){
								String count = redisService.get(logKey);
								redisService.set(logKey, Integer.parseInt(count)+1+"", 2*24*60*60*1000l);
							}else{
								redisService.set(logKey, "1", 2*24*60*60*1000l);
							}
						}
						if(buff.size()>0){
//							producer.send(new ProducerRecord<String, String>(topic +"_etl", encrypt(nginxInfo, fieldTypes, buff)));
						}
					}
					consumer.commitSync();
				} catch (Exception e) {
					e.printStackTrace();
					LOG.error("拉取用户行为数据失败:"+e);
				}
			}
		}finally {
			try {
				if (httpclient != null) {
					httpclient.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
	
	
	private String encrypt(String nginxInfo,Map<String,String> fieldTypes, List<RecordItem> buff) {
		List<Map<String, Object>> maps = new ArrayList<Map<String, Object>>();
		for(RecordItem item : buff){
			HashMap<String,Object> map = JSONObject.parseObject(JSONObject.toJSONString(item), HashMap.class);
			map.putAll(item.getExtras());
			maps.add(map);
		}
		
		Set<String> fields = new HashSet<String>();
		for(Map<String, Object> map : maps){
			fields.addAll(map.keySet());
		}
		fields.remove("id");
		fields.remove("extras");
		fields.remove("ignores");
		fields.remove("errorType");
		fields.remove("osplat");
		fields.remove("ispro");
		StringBuffer headStr = new StringBuffer();
		Map<String,StringBuffer> bodys = new HashMap<String,StringBuffer>();
		for(String field: fields){
			String type=(fieldTypes.get(field)!=null?fieldTypes.get(field):"s");
			if(redisService.exists(MatcherConfig.FIELD_TYPE+ field)){
				type = redisService.get(MatcherConfig.FIELD_TYPE+ field);
			}
			headStr.append(type+"|"+field+",");
			
			for( int i=0 ; i< maps.size(); i++){
				StringBuffer body =bodys.get(i+"");
				if(body==null){
					body =new StringBuffer();
				}
				body.append(maps.get(i).get(field)+ RecordItem.ITEM_TERMINATED);
				bodys.put(i+"", body);
			}
		}
		StringBuffer bodyStr =  new StringBuffer();
		for(StringBuffer body: bodys.values()){
			bodyStr.append(body.substring(0, body.length()-1)+RecordItem.COLLECTION_TERMINATED);
		}
		return nginxInfo +Base64Coder.encodeString( headStr.substring(0, headStr.length()-1) +RecordItem.COLLECTION_TERMINATED+bodyStr.substring(0, bodyStr.length()-1));
	}


	/**
	 * 优化策略，如果缓存里面不存在且value出现一次以上才请求远程库，当请求两次还是没有数据，则三个小时内忽略此value
	 * @param value 事件值
	 * @return
	 */
	private boolean accumulate(String type, String eventValue){
		if(redisService.exists(type+matcherConfig.Ignore+eventValue)){
			int cnt = Integer.parseInt(redisService.get(type+matcherConfig.Ignore+eventValue));
			if(cnt < 3){
				redisService.set(type+matcherConfig.Ignore+eventValue, cnt+1+"", 6*60*60*1000l);
				return true;
			}
		}else{
			redisService.set(type+matcherConfig.Ignore+eventValue, "1", 6*60*60*1000l);
		}
		return false;
	}
	
	/**
	 * 添加各种字段的处理
	 * @param one
	 * @throws Exception
	 */
	public void dataHandle(CloseableHttpClient httpclient, RecordItem one, String osplat) throws Exception{
		//抽取手机号
		String userID = one.getUserID();
		if(userID != null && userID.matches("^\\d{11}$")){
			one.setUserID4Num(userID);
		}
		//菜单相关
		/*String menu = one.getEvent_name();
		if(menu != null && !"".equals(menu.trim()) ){
			try{
				//分类
				if(redisService.exists(MatcherConfig.MENU_OPT_RELATE+menu)){
						one.setOptName(redisService.get(menu));
				}
				//场景
				if(redisService.exists(MatcherConfig.MENU_SCENE_RELATE+menu)){
					one.setSceneName(redisService.get(MatcherConfig.MENU_SCENE_RELATE+menu));
				}
				//菜单级别
				if(redisService.exists(MatcherConfig.MENU_LEVEL+menu)){
					one.setMenuLevel(redisService.get(MatcherConfig.MENU_LEVEL+menu));
				}
			}catch(Exception e){
				System.out.println(menu);
			}
		}*/
		//日期对应周(非自然周)
		String date = one.getEventDate();
		if(date != null && !"".equals(date.trim())){
			if(redisService.exists(date)){
				one.setYearWeek(redisService.get(date));
			}
		}
		//根据event_value填充不同内容
		String eventValue = one.getEventValue();
		if(eventValue !=null && eventValue.matches("^[0-9a-zA-Z\\-]+$")){
			String eventValueType = one.getEventValueType();
			/*
			产品
			文章
			课程
			订单
			组合类别
			*/
			
			//产品类型
			if("pid".equalsIgnoreCase(eventValueType) && eventValue.matches("^[0-9\\-]{8}$")){
				if(redisService.exists(matcherConfig.getProductTitle()+eventValue) ){// || (accumulate(MatcherConfig.Product, eventValue) && loadDataTask.queryProduct(eventValue))){
					one.setContent(redisService.get(matcherConfig.getProductTitle()+eventValue));
				}
			//文章资讯类型
			}else if("news".equalsIgnoreCase(eventValueType) && eventValue.matches("^\\d{1,5}$")){
				if(redisService.exists(MatcherConfig.TITLE+eventValue)){
					one.setContent(redisService.get(MatcherConfig.TITLE+eventValue));
					if(redisService.exists(MatcherConfig.NAME+eventValue)){
						one.setColumnName(redisService.get(MatcherConfig.NAME+eventValue));
					}
				}else if(redisService.exists(matcherConfig.getContentTitle()+eventValue) ){//||  (accumulate(MatcherConfig.Content, eventValue) && loadDataTask.queryContent(eventValue)) ){
					one.setContent(redisService.get(matcherConfig.getContentTitle()+eventValue));
					if(redisService.exists(matcherConfig.getContentName()+eventValue)){
						one.setColumnName(redisService.get(matcherConfig.getContentName()+eventValue));
					}
				}
			}
		}
		
		String deviceId = one.getDevice_id();
		String newDeviceSign = null;
		if("app".equals(osplat)){
			//启动标记
			String eventAction = one.getEvent_type();
			if("启动".equals(eventAction) || "唤醒".equals(eventAction)){
//			kafkaLogService.addStartupLog(new StartupLog(one.getClientDeviceID(), one.getEventDateTime() , new Date()));
				String key = MatcherConfig.DEVICE_STARTUP+deviceId;
				long newTime = Long.parseLong(one.getEvent_time());
				if(!redisService.exists(key)){
					one.setIsStartup("1");
					redisService.sadd(key, newTime+"");
					redisService.expire(key, MatcherConfig.OVERTIME_4_REDIS);
				}else {
					Set<Object> cacheTimes = redisService.smembers(key);
					boolean isStartup = true;
					for(Object ct : cacheTimes){
						long cacheTime = Long.parseLong(ct+"");
						long diff = cacheTime - newTime;
						if(diff>-30000 && diff<30000){
							isStartup = false;
							break;
						}
					}
					if(isStartup){
						one.setIsStartup("1");
						redisService.sadd(key, newTime+"");
					}
					redisService.expire(key, MatcherConfig.OVERTIME_4_REDIS);
				}
			}
			newDeviceSign = MatcherConfig.EXISTS_DEVICEID+deviceId;
		}else{
			newDeviceSign = MatcherConfig.EXISTS_DEVICEID+osplat+one.getDevice_id();
		}
		
		if(redisService.exists(newDeviceSign)){
			redisService.expire(newDeviceSign, 365*24*60*60*1000l);
		}else{
			one.setIsNewUser("1");
			redisService.set(newDeviceSign, "");
		}
		
		//是否第一次登陆卡号
		String newUserIDSign = MatcherConfig.EXISTS_USERID + one.getUserID();
		if(!redisService.exists(newUserIDSign)){
			one.setIsFirstLogin("1");
			one.setChannelFirstLogin("1");
			redisService.set(newUserIDSign, osplat);
			redisService.expire(newUserIDSign, 365*24*60*60*1000l);
		}else{
			String mark = redisService.get(newUserIDSign);
			if(!"all".equals(mark) && !osplat.equals(mark)){
				one.setChannelFirstLogin("1");
				redisService.set(newUserIDSign, "all");
				redisService.expire(newUserIDSign, 365*24*60*60*1000l);
			}
		}
		
//		one.setIsValidData("Y");
		//当前消息放入到队列中
//		msgQueue.offer(one);
	}
	

	/**
	 * 创建kafka消费端，设置手动commit
	 * @param topic
	 * @return
	 */
	private KafkaConsumer<String, String> createConsumer(String topic) {
		Properties props = kafkaConfig.getProperties(true);
		LOG.info("kafka config:" + props);
		KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
		
		List<TopicPartition> topicPartitions = new ArrayList<TopicPartition>();
        List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);
        for (PartitionInfo partition: partitionInfos) {
            topicPartitions.add(new TopicPartition(topic, partition.partition()));
        }
        consumer.assign(topicPartitions);
        
        for (TopicPartition topicPartition: topicPartitions) {
        	//初始化offset位置，只有第一次运行时为true
            if (consumer.committed(topicPartition) == null) {
                long position = consumer.position(topicPartition);
                LOG.info("topic partition " + topicPartition.topic()+":"+topicPartition.partition() + ", offset: " + position);
                kafkaLogService.addStartupLog(new KafkaStartupLog(topic, ""+topicPartition.partition(), ""+position));
                consumer.seek(topicPartition, position);
            }
        }
		return consumer;
	}
public static void main(String[] args) {
	List<String> kk = null;
	for(String s: kk){
		System.out.println(s);
	}
//	System.out.println(Base64Coder.decodeString("c3xldmVudF9uYW1lLHN8c3lzdGVtX25hbWUsc3xicm93c2VyLHN8cmVmZXJyZXIsc3xyZWZlcnJpbmdfZG9tYWluLHN8ZXZlbnRfdGltZSxzfGN1cnJlbnRfdXJsLHN8YnJvd3Nlcl92ZXJzaW9uLGx8c2NyZWVuX2hlaWdodCxsfHNjcmVlbl93aWR0aCxzfHN1Z29fbGliLHN8c2RrX3ZlcnNpb24sc3xkaXN0aW5jdF9pZCxzfENTMSxzfHNlc3Npb25faWQsc3xhcHBfbmFtZSxzfGFwcF92ZXJzaW9uLHN8VXNlcklELHN8cGFnZV9uYW1lLHN8ZXZlbnRfdHlwZSxzfGhvc3Qsc3xwYXRoX25hbWUsc3x0b2tlbgLmtY/op4gBV2luZG93cwFDaHJvbWUBZGlyZWN0AWRpcmVjdAExNDg3ODMzMDUzMjE2AWh0dHBzOi8vZWxuLXRlc3QuaW5maW5pdHVzLmNvbS5jbi9lbG4tZnJvbnQvZnJvbnQvY291cnNlY2VudGVyL2RldGFpbD9jb3Vyc2VpZD0xNTMBNTUBNzIwATEyODABd2ViATAuMS4zATE1YTVmZjJlNGY1NGY4LTA0ZDI4MThiNzkzNDcyLTVkNGUyMTFmLWUxMDAwLTE1YTVmZjJlNGY2ODkxATA0MTAwMjYxMAE3MDMyNTdkYi1iNmVlLTQ1N2QtODc5OC03MDI4MGVkOTA1MzEB5Lia5Yqh57uf5LiA5bmz5Y+wUEMgATEuMS4xATA0MTAwMjYxMAHor77nqIvkuK3lv4MB5rWP6KeIAWVsbi10ZXN0LmluZmluaXR1cy5jb20uY24BL2Vsbi1mcm9udC9mcm9udC9jb3Vyc2VjZW50ZXIvZGV0YWlsATRmMDkyNjA5OTBhYWFmNGNhMzI4YTU4NWI2MTM4YzBi"));
}
}
